import datetime
import threading
from concurrent.futures._base import wait, ALL_COMPLETED
from concurrent.futures.thread import ThreadPoolExecutor
from queue import Queue

import schedule
from sqlalchemy.orm import sessionmaker

from IntelligentDocumentReview.DbInit.mysqlDb import session, engine
from IntelligentDocumentReview.Model.Message import Messages
from IntelligentDocumentReview.Model.Order import Orders

import asyncio

def replyDetection():
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func time :',ts)

    try:
        orderlist = session.execute('select * from orders where reviewType = 2 and replyStatus in(1, 2) and endReplyTime <= now()').fetchall()
        for order in orderlist:
            if order.replyStatus == 1:
                session.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 2, "endReplyTime": order.endReplyTime+datetime.timedelta(hours=12)})

                m = Messages()
                m.orderId = order.id
                m.senderId = 0
                m.senderName = "admin"
                m.receiverId = order.nextPersonId
                m.receiverName = order.nextPersonName
                m.status = 1
                session.add(m)

            elif order.replyStatus == 2:
                session.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 3, "status": 1})

        session.commit()

    except Exception:
        raise
        session.rollback()

def replyDetectionAsyncNoLock(i, sessionAsync):
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func time :',ts)

    try:
        orderlist = sessionAsync.execute('select * from orders where reviewType = 2 and replyStatus in(1, 2) and endReplyTime <= now() limit 3 offset :page', params={"page":i * 3}).fetchall()
        for order in orderlist:
            if order.replyStatus == 1:
                sessionAsync.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 2, "endReplyTime": order.endReplyTime+datetime.timedelta(hours=12)})

                m = Messages()
                m.orderId = order.id
                m.senderId = 0
                m.senderName = "admin"
                m.receiverId = order.nextPersonId
                m.receiverName = order.nextPersonName
                m.status = 1
                sessionAsync.add(m)

            elif order.replyStatus == 2:
                sessionAsync.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 3, "status": 1})

        sessionAsync.commit()

    except Exception:
        session.rollback()

        raise

def replyDetectionAsync(i, lock):
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func time :',ts)

    lock.acquire()

    try:
        orderlist = session.execute('select * from orders where reviewType = 2 and replyStatus in(1, 2) and endReplyTime <= now() limit 3 offset :page', params={"page":i * 3}).fetchall()
        for order in orderlist:
            if order.replyStatus == 1:
                session.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 2, "endReplyTime": order.endReplyTime+datetime.timedelta(hours=12)})

                m = Messages()
                m.orderId = order.id
                m.senderId = 0
                m.senderName = "admin"
                m.receiverId = order.nextPersonId
                m.receiverName = order.nextPersonName
                m.status = 1
                session.add(m)

            elif order.replyStatus == 2:
                session.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 3, "status": 1})

        session.commit()

    except Exception as e:
        session.rollback()
        print("error：", e)

        lock.release()
        raise

    print("结束。。。。。。。。。。。。。。。。。。")
    lock.release()

def replyDeteyionAsync():
    task_list = []
    thread_pool = ThreadPoolExecutor(5)
    lock = threading.Lock()
    for i in range(0 ,5):
        o = i
        # task_list.append(thread_pool.submit(replyDetectionAsync, o, lock))

        sessionAsync = sessionmaker(bind=engine)()
        task_list.append(thread_pool.submit(replyDetectionAsyncNoLock, o, sessionAsync))

    wait(task_list, return_when=ALL_COMPLETED)
    thread_pool.shutdown()

    print("当次任务结束。。。。。。。。。。。。。。。。。。")

async def replyDetectionPrimordialAsync(i, sessionAsync):
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func time :',ts)

    try:
        orderlist = sessionAsync.execute('select * from orders where reviewType = 2 and replyStatus in(1, 2) and endReplyTime <= now() limit 3 offset :page', params={"page":i * 3}).fetchall()
        for order in orderlist:
            if order.replyStatus == 1:
                sessionAsync.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 2, "endReplyTime": order.endReplyTime+datetime.timedelta(hours=12)})

                m = Messages()
                m.orderId = order.id
                m.senderId = 0
                m.senderName = "admin"
                m.receiverId = order.nextPersonId
                m.receiverName = order.nextPersonName
                m.status = 1
                sessionAsync.add(m)

            elif order.replyStatus == 2:
                sessionAsync.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 3, "status": 1})

        sessionAsync.commit()

    except Exception as e:
        print("e：", e)
        sessionAsync.rollback()

        raise

    print("结束。。。。。。。。。。。。。。。。。。")

async def replyDeteyionPrimordialAsync():
    task_list = []

    # new_loop = asyncio.new_event_loop()
    # asyncio.set_event_loop(new_loop)

    for i in range(0, 5):
        o = i
        sessionAsync = sessionmaker(bind=engine)()
        cor = replyDetectionPrimordialAsync(o, sessionAsync)
        # task = asyncio.ensure_future(cor)
        task = asyncio.create_task(cor)
        task_list.append(task)

    # loop = asyncio.get_event_loop()
    # loop.run_until_complete(asyncio.wait(task_list))

    await asyncio.wait(task_list)

    print("当次任务结束。。。。。。。。。。。。。。。。。。")

def startAsync():
    asyncio.run(replyDeteyionPrimordialAsync())

def replyDetectionQueue(q, sessionAsync):
    while not q.empty():
        order = q.get()
        if order.replyStatus == 1:
            sessionAsync.query(Orders).filter(Orders.id == order.id).update(
                {"replyStatus": 2, "endReplyTime": order.endReplyTime + datetime.timedelta(hours=12)})
            sessionAsync.commit()

            m = Messages()
            m.orderId = order.id
            m.senderId = 0
            m.senderName = "admin"
            m.receiverId = order.nextPersonId
            m.receiverName = order.nextPersonName
            m.status = 1
            sessionAsync.add(m)
            sessionAsync.commit()

        elif order.replyStatus == 2:
            sessionAsync.query(Orders).filter(Orders.id == order.id).update({"replyStatus": 3, "status": 1})
            sessionAsync.commit()

    print("结束。。。。。。。。。。。。。。。。。。")

def replyDetectionQueueAsync():
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func time :',ts)

    q = Queue(maxsize=15)
    try:
        orderlist = session.execute('select * from orders where reviewType = 2 and replyStatus in(1, 2) and endReplyTime <= now()').fetchall()
        print("len：", len(orderlist))
        for order in orderlist:
            q.put(order)

        task_list = []

        for i in range(5):
            sessionAsync = sessionmaker(bind=engine)()

            t = threading.Thread(target=replyDetectionQueue, args=(q,sessionAsync,))
            # t.setDaemon(True)
            t.start()
            task_list.append(t)

        for task in task_list:
            task.join()

    except Exception:
        session.rollback()

        raise

    print("当次任务结束。。。。。。。。。。。。。。。。。。")

def replyDeteyionThread():
    task_list = []

    lock = threading.Lock()
    for i in range(0, 5):
        o = i
        # t = threading.Thread(target=replyDetectionAsync, args=(o,lock,))

        sessionAsync = sessionmaker(bind=engine)()
        t = threading.Thread(target=replyDetectionAsyncNoLock, args=(o, sessionAsync,))

        t.start()
        task_list.append(t)

    for task in task_list:
        task.join()

    print("当次任务结束。。。。。。。。。。。。。。。。。。")

def tasklist():

    #清空任务
    schedule.clear()

    #创建一个2分钟间隔执行任务
    # schedule.every(1).minutes.do(replyDetection)
    # schedule.every(1).minutes.do(replyDeteyionAsync)
    schedule.every(1).minutes.do(startAsync)
    # schedule.every(1).minutes.do(replyDeteyionThread)
    # schedule.every(1).minutes.do(replyDetectionQueueAsync)

    while True:
        schedule.run_pending()


def reply(orderId, result):

    if result == "1":
        replyStatus = 4
    else:
        replyStatus = 5

    orderStatus = 0
    if replyStatus == 4:
        orderStatus = 5
    elif replyStatus == 5:
        orderStatus = 1

    try:
        session.query(Orders).filter(Orders.id == int(orderId)).update({"replyStatus": replyStatus, "status": orderStatus})

        session.commit()
    except:
        raise
        session.rollback()

    return {"code":200, "data":None, "message":"回复成功"}

def insertOrder(o):
    print("o：",
        o.customerId,
        o.customerName,
        o.nextPersonId,
        o.nextPersonName,
        o.amountMoney,
        o.commodityListStr,
          )

    try:
        session.add(o)

        session.commit()
    except:
        session.rollback()

    return {"code": 200, "data": None, "message": "下单成功"}

def selectReviewType(t, orderId):

    if t == "2":

        try:
            session.query(Orders).filter(Orders.id == orderId).update({"reviewType": 2, "replyStatus":1, "endReplyTime":datetime.datetime.now()+datetime.timedelta(hours=12)})
            session.commit()
        except:
            session.rollback()

    return {"code": 200, "data": None, "message": "选择成功"}












